package jast.util.es;


import lombok.extern.slf4j.Slf4j;
import lombok.extern.slf4j.XSlf4j;
import org.elasticsearch.action.bulk.BulkItemResponse;
import org.elasticsearch.action.bulk.BulkRequestBuilder;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteRequestBuilder;
import org.elasticsearch.action.support.WriteRequest;
import org.elasticsearch.action.update.UpdateRequestBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * ES 操作类
 */
public class ElasticSearchBulkOperator {

    private static final Logger LOG = LoggerFactory.getLogger(ElasticSearchBulkOperator.class);

//    private static Log LOG = LogFactory.get(wiki.hadoop.article.util.ElasticSearchBulkOperator.class);


    private static final int MAX_BULK_COUNT = 10000;

    private static BulkRequestBuilder bulkRequestBuilder = null;

    private static final Lock commitLock = new ReentrantLock();

    private static ScheduledExecutorService scheduledExecutorService = null;

    static {
        init();
    }

    public static void init() {
        // 初始化  bulkRequestBuilder
        bulkRequestBuilder = ElasticSearchClientManager.getESClient().prepareBulk();
        bulkRequestBuilder.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE);

        // 初始化线程池大小为1
        scheduledExecutorService = Executors.newScheduledThreadPool(1);

        //创建一个Runnable对象，提交待写入的数据，并使用commitLock锁保证线程安全
        final Runnable beeper = () -> {
            commitLock.lock();
            try {
                LOG.info("Before submission bulkRequest size : " + bulkRequestBuilder.numberOfActions());
                //提交数据至es
                bulkRequest(0);
                LOG.info("After submission bulkRequest size : " + bulkRequestBuilder.numberOfActions());
            } catch (Exception ex) {
                ex.printStackTrace();
                LOG.error(ex.getMessage());
            } finally {
                commitLock.unlock();
            }
        };

        //初始化延迟10s执行 runnable方法，后期每隔30s执行一次
        scheduledExecutorService.scheduleAtFixedRate(beeper, 10, 30, TimeUnit.SECONDS);
    }

    public static void shutdownScheduEx() {
        if (null != scheduledExecutorService && !scheduledExecutorService.isShutdown()) {
            scheduledExecutorService.shutdown();
        }
    }

    /**
     * 批量提交
     *
     * @param threshold 待提交数量大于threshold 数量则进行提交
     */
    public static void bulkRequest(int threshold) {
//        synchronized (bulkRequestBuilder) {
            //如果待提交数量大于传入的threshold数量则提交
            if (bulkRequestBuilder.numberOfActions() > threshold) {
                BulkResponse bulkItemResponse = bulkRequestBuilder.execute().actionGet();
                if (!bulkItemResponse.hasFailures()) {
                    int beforeCount = bulkRequestBuilder.numberOfActions();
                    bulkRequestBuilder = ElasticSearchClientManager.getESClient().prepareBulk();
                    LOG.info("提交成功,提交前" + beforeCount + "\t提交后:" + bulkRequestBuilder.numberOfActions());
                } else {
                    errorDataProcess(bulkItemResponse);
                }
            }
//        }
    }

    /**
     * 提交异常的数据处理
     *
     * @param bulkItemResponse
     */
    private static void errorDataProcess(BulkResponse bulkItemResponse) {
        LOG.error("提交请求失败，异常信息为:" + bulkItemResponse.buildFailureMessage());
        BulkItemResponse[] items = bulkItemResponse.getItems();
        for (BulkItemResponse item : items) {
            //是否导入失败
            if (item.isFailed()) {
                //导入失败数据的id
                String failId = item.getId();
                //导入失败详情
                BulkItemResponse.Failure failure = item.getFailure();
                //导入失败信息
                String failureMessage = item.getFailureMessage();
                System.out.println(item.isFailed() + "\t" + item.getFailureMessage() + "\t" + item.getFailure() + "\t" + item.getId());
                //TODO 异常信息处理逻辑
            }
        }
        //处理完成初始化 bulkRequestBuilder 对象
        bulkRequestBuilder =  ElasticSearchClientManager.getESClient().prepareBulk();

    }

    /**
     * add update builder to bulk
     * use commitLock to protected bulk as thread-save
     *
     * @param builder
     */

    public static void addUpdateBuilderToBulk(UpdateRequestBuilder builder) {
        commitLock.lock();
        try {
            bulkRequestBuilder.add(builder);
            bulkRequest(MAX_BULK_COUNT);
        } catch (Exception ex) {
            LOG.error(" update Bulk index error : " + ex.getMessage() + "\t" + ex.toString());
            ex.printStackTrace();
        } finally {
            commitLock.unlock();
        }
    }

    /**
     * add delete builder to bulk
     * use commitLock to protected bulk as thread-save
     *
     * @param builder
     */
    public static void addDeleteBuilderToBulk(DeleteRequestBuilder builder) {
        commitLock.lock();
        try {
            bulkRequestBuilder.add(builder);
            bulkRequest(MAX_BULK_COUNT);
        } catch (Exception ex) {
            LOG.error(" delete Bulk " + "gejx_test" + " index error : " + ex.getMessage());
        } finally {
            commitLock.unlock();
        }
    }


}
